package com.yinuo.starter.factory;

import com.alibaba.fastjson.JSONObject;
import com.yinuo.starter.anno.ConsumerProcessor;
import com.yinuo.starter.bean.TransactionalBean;
import com.yinuo.starter.config.TransactionalInit;
import com.yinuo.starter.dao.TransactionalDao;
import com.yinuo.starter.properties.ConsumerProperties;
import com.yinuo.starter.properties.JmsProperties;
import com.yinuo.starter.result.ResultBase;
import com.yinuo.starter.util.Str;
import com.yinuo.starter.util.TransactionalMessage;
import com.yinuo.starter.util.SpringUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.transaction.annotation.Transactional;

import java.sql.Timestamp;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentMap;

@Slf4j
public class ConsumerFactory {

    private TransactionalDao transactionalDao;


    public ConsumerFactory(JmsProperties jmsProperties, ConsumerProperties consumerProperties,
                           ConcurrentMap<String, ConsumerProcessor.ListenerMethod> methodCache,
                           TransactionalInit transactionalInit, TransactionalDao transactionalDao) {
        this.transactionalDao = transactionalDao;
        try{
            transactionalInit.tableHandler();
            ConsumerFactory(jmsProperties, consumerProperties, methodCache);
        }catch (Exception e){
            e.printStackTrace();
        }
    }

    private void ConsumerFactory(JmsProperties jmsProperties, ConsumerProperties consumerProperties
            ,ConcurrentMap<String, ConsumerProcessor.ListenerMethod> methodMap) throws MQClientException {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(consumerProperties.getGroupName());
        consumer.setNamesrvAddr(Str.checkAddr(jmsProperties.getNameSrvAddr()));
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        consumer.setConsumeMessageBatchMaxSize(consumerProperties.getConsumeMessageBatchMaxSize());
        consumer.setMaxReconsumeTimes(consumerProperties.getMaxReconsumeTimes());
        for (Map.Entry<String, ConsumerProcessor.ListenerMethod> entry : methodMap.entrySet()) {
            consumer.subscribe(entry.getValue().annotations.topicName(), "*");
        }
        consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> handler(msgs, methodMap));
        consumer.start();
    }

    private TransactionalBean getTransaction(MessageExt message){
        TransactionalBean transactionalBean = transactionalDao.selectByPrimary("C" + message.getKeys(), 2);
        return transactionalBean;
    }

    @Transactional(rollbackFor = Exception.class)
    public ConsumeConcurrentlyStatus handler(List<MessageExt> msgs, ConcurrentMap<String, ConsumerProcessor.ListenerMethod> methodMap){
        MessageExt msg = msgs.get(0);
        String message = new String(msgs.get(0).getBody());
        TransactionalMessage transactionalMessage = JSONObject.parseObject(message, TransactionalMessage.class);
        StringBuffer sb = new StringBuffer().append(msg.getTopic());
        if(msg.getTags() != null){
            sb.append(msg.getTags());
        }
        ConsumerProcessor.ListenerMethod listenerMethod = methodMap.get(sb.toString());
        Integer r;
        try {
            TransactionalBean transactionalBean = getTransaction(msg);
            if(transactionalBean != null){
                if(Integer.valueOf(transactionalBean.getValue()) == ResultBase.CONSUMER_SUCCESS){
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
                transactionalDao.deleteByPrimary(transactionalBean.getId(), transactionalBean.getType());
            }

            Object o = SpringUtil.getBean(listenerMethod.targetClass);
            r = (Integer) o.getClass().getMethod(listenerMethod.method.getName(), TransactionalMessage.class)
                    .invoke(o, transactionalMessage);
            transactionalBean = new TransactionalBean();
            transactionalBean.setId("C" + msg.getKeys());
            transactionalBean.setTopicName(msg.getTopic());
            transactionalBean.setTagName(msg.getTags());
            transactionalBean.setCreateTime(new Timestamp(new Date().getTime()));
            transactionalBean.setModifyTime(transactionalBean.getCreateTime());
            transactionalBean.setKeys(msg.getKeys());
            transactionalBean.setType(2);
            transactionalBean.setValue(r == null || r == ResultBase.CONSUMER_SUCCESS
                    ? String.valueOf(ResultBase.CONSUMER_SUCCESS) : String.valueOf(ResultBase.CONSUMER_FAIL));
            if(r == null){
                r = ResultBase.CONSUMER_SUCCESS;
            }
            transactionalDao.insert(transactionalBean);
        } catch (Exception e){
            log.error(e.getMessage());
            e.printStackTrace();
            return consumerMessageHandle(msg, ResultBase.CONSUMER_FAIL);
        }
        return consumerMessageHandle(msg, r);
    }

    private ConsumeConcurrentlyStatus consumerMessageHandle(MessageExt message, Integer ret){
        if(ret == null || ret == ResultBase.CONSUMER_SUCCESS){
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        }
        return ConsumeConcurrentlyStatus.RECONSUME_LATER;
    }

}
